package com.huatai.datacenter.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.annotation.TableName;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.huatai.datacenter.entity.TbQualityRunLogEntity;
import com.huatai.datacenter.entity.TopicTaskEntity;
import com.huatai.datacenter.entity.datamodel.DataModelEntity;
import com.huatai.datacenter.entity.datamodel.DataModelFieldEntity;
import com.huatai.datacenter.entity.dataquality.QualityLogDetailEntity;
import com.huatai.datacenter.entity.dataquality.QualityLogEntity;
import com.huatai.datacenter.entity.dataquality.QualityRuleEntity;
import com.huatai.datacenter.entity.postdata.HttpRequestEntity;
import com.huatai.datacenter.mapper.QualityLogDetailMapper;
import com.huatai.datacenter.mapper.QualityLogMapper;
import com.huatai.datacenter.mapper.QualityRuleMapper;
import com.huatai.datacenter.mapper.TbQualityRunLogMapper;
import com.huatai.datacenter.service.impl.QueryServiceImpl;
import com.huatai.dataservice.dubbo.DataDisposeDubboAPI;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.BufferedReader;
import java.io.DataOutputStream;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URL;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author Lion
 * @date 2023/6/7  15:42
 */
@Slf4j
@Component
public class ConsumerListener {

	@Autowired
	private HttpRequestService httpRequestService;

	@Autowired
	private DataModelFieldService fieldService;

	@DubboReference
	private DataDisposeDubboAPI dataDisposeDubbo;

	@Autowired
	private QueryServiceImpl queryServiceImpl;
	@Autowired
	private QualityLogMapper qualityLogMapper;

	@Autowired
	private QualityLogDetailMapper qualityLogDetailMapper;

	@Autowired
	private QualityRuleMapper  qualityRuleMapper;

	//@XxlJob("demoJobHandler")
	public void demoJobHandler() throws Exception {
		XxlJobHelper.log("XXL-JOB, Hello World.");

		for (int i = 0; i < 5; i++) {
			XxlJobHelper.log("beat at:" + i);
			TimeUnit.SECONDS.sleep(2);
		}
		// default success
	}

	@Autowired
	private TbQualityRunLogMapper tbQualityRunLogMapper;

	/**
	 * 4、跨平台Http任务
	 */
	@XxlJob("httpJobHandler")
	public void httpJobHandler() throws Exception {

		//获取参数
		String param = XxlJobHelper.getJobParam();
		if (param==null || param.trim().length()==0) {
			XxlJobHelper.log("param["+ param +"] invalid.");
			XxlJobHelper.handleFail();
			return;
		}

		JSONObject jsonObject = JSONObject.parseObject(param);
		String url = jsonObject.getString("url");
		String method = jsonObject.getString("method");
		String data = jsonObject.getString("data");
		long id = jsonObject.getLong("id");

		if (url==null || url.trim().length()==0) {
			XxlJobHelper.log("url["+ url +"] invalid.");
			XxlJobHelper.handleFail();
			return;
		}
		if (method==null || !Arrays.asList("GET", "POST").contains(method)) {
			XxlJobHelper.log("method["+ method +"] invalid.");
			XxlJobHelper.handleFail();
			return;
		}
		boolean isPostMethod = method.equals("POST");

		// request
		HttpURLConnection connection = null;
		BufferedReader bufferedReader = null;
		try {
			// connection
			URL realUrl = new URL(url);
			connection = (HttpURLConnection) realUrl.openConnection();

			// connection setting
			connection.setRequestMethod(method);
			connection.setDoOutput(isPostMethod);
			connection.setDoInput(true);
			connection.setUseCaches(false);
			connection.setReadTimeout(5 * 1000);
			connection.setConnectTimeout(3 * 1000);
			connection.setRequestProperty("connection", "Keep-Alive");
			connection.setRequestProperty("Content-Type", "application/json;charset=UTF-8");
			connection.setRequestProperty("Accept-Charset", "application/json;charset=UTF-8");

			// do connection
			connection.connect();

			// data
			if (isPostMethod && data!=null && data.trim().length()>0) {
				DataOutputStream dataOutputStream = new DataOutputStream(connection.getOutputStream());
				dataOutputStream.write(data.getBytes("UTF-8"));
				dataOutputStream.flush();
				dataOutputStream.close();
			}

			// valid StatusCode
			int statusCode = connection.getResponseCode();
			if (statusCode != 200) {
				throw new RuntimeException("Http Request StatusCode(" + statusCode + ") Invalid.");
			}

			// result
			bufferedReader = new BufferedReader(new InputStreamReader(connection.getInputStream(), "UTF-8"));
			StringBuilder result = new StringBuilder();
			String line;
			while ((line = bufferedReader.readLine()) != null) {
				result.append(line);
			}
			//添加查询方法 apiAddress requestType
			LambdaQueryWrapper<HttpRequestEntity> wrapper = Wrappers.lambdaQuery();
			wrapper.eq(HttpRequestEntity::getApiAddress,url).eq(HttpRequestEntity::getRequestType,method);
//			QueryWrapper<HttpRequestEntity> queryWrapper = new QueryWrapper<>();
//			queryWrapper.eq("id",1933102486314242049L);

			HttpRequestEntity httpRequest = httpRequestService.getOne(wrapper);

			TbQualityRunLogEntity qualityrunLogEntity = new TbQualityRunLogEntity();
			//http运行日志类型 begin  dataCount  end
			qualityrunLogEntity.setRequestId(String.valueOf(httpRequest.getId()));
			qualityrunLogEntity.setCreateTime(LocalDateTime.now());
			qualityrunLogEntity.setTables(httpRequest.getTableName());
			qualityrunLogEntity.setLogType("begin");
			qualityrunLogEntity.setRemark(LocalDateTime.now() + "采集开始");
			qualityrunLogEntity.setDeleted(0);
			tbQualityRunLogMapper.insert(qualityrunLogEntity);

			TbQualityRunLogEntity LogEntity = new TbQualityRunLogEntity();

			//获取指定结构
			JSONArray dataArray = null;
			if ("data".equals(httpRequest.getStructure())){
				dataArray = JSON.parseArray(result.toString());
				LogEntity.setRemark(LocalDateTime.now() + "入库" + dataArray.size() + "条数据");

			}else {
				dataArray = queryJson(JSONObject.parseObject(result.toString()),httpRequest.getStructure());
				String dataCount = result.toString();
				ObjectMapper mapper = new ObjectMapper();
				JsonNode root = mapper.readTree(dataCount);
				//统计条数
				int count =root.path("data").path("records").size();
				LogEntity.setRemark(LocalDateTime.now() + "入库" + count + "条数据");
			}
			LogEntity.setRequestId(String.valueOf(httpRequest.getId()));
			LogEntity.setCreateTime(LocalDateTime.now());
			LogEntity.setTables(httpRequest.getTableName());
			LogEntity.setLogType("dataCount");
			LogEntity.setDeleted(0);
			tbQualityRunLogMapper.insert(LogEntity);

			TbQualityRunLogEntity RunLog = new TbQualityRunLogEntity();
			RunLog.setRequestId(String.valueOf(httpRequest.getId()));
			RunLog.setCreateTime(LocalDateTime.now());
			RunLog.setTables(httpRequest.getTableName());
			RunLog.setLogType("end");
			RunLog.setRemark(LocalDateTime.now() + "采集结束");
			RunLog.setDeleted(0);
			tbQualityRunLogMapper.insert(RunLog);


			//获取字段
			List<DataModelFieldEntity> dataModelFieldEntities = fieldService.getDataModelFieldList(httpRequest.getModelId());


			String modeName = queryServiceImpl.saveQualityLog(httpRequest.getModelId());

			QualityLogEntity qualityLogEntity = new QualityLogEntity();
			qualityLogEntity.setDataModelId(httpRequest.getModelId());
			qualityLogEntity.setDataModelName(String.valueOf(modeName));
			qualityLogEntity.setLastExecuteTime(LocalDateTime.now());
			qualityLogMapper.insert(qualityLogEntity);

			// 创建用于保存详情记录的列表
			List<QualityLogDetailEntity> nullValueDetails = new ArrayList<>();
			String ruleName = "完整性";
			Long qualityRuleId = qualityRuleMapper.selectId(ruleName);

			// 遍历 dataArray 中的每个对象
			for (int i = 0; i < dataArray.size(); i++) {
				JSONObject record = dataArray.getJSONObject(i);
				// 遍历记录中的所有字段
				for (String fieldName : record.keySet()) {
					Object value = record.get(fieldName);
					// 检查字段值是否为 null
					if (value == null) {
						QualityLogDetailEntity detail = new QualityLogDetailEntity();
						detail.setQualityLogId(qualityLogEntity.getId());
						detail.setFieldName(fieldName);
						detail.setQualityRuleId(qualityRuleId);
						detail.setQualityRuleName("完整性");
						detail.setRemark("内部方法API(null值检测):nullDetection校验失败，原因：源数据值为空");
						detail.setExecuteTime(LocalDateTime.now());
						qualityLogDetailMapper.insert(detail);
						// 设置源数据为当前记录的ID（如果有）
						String recordId = record.getString("ID");
						if (recordId != null) {
							detail.setSourceData("记录ID: " + recordId);
						} else {
							detail.setSourceData("记录索引: " + i);
						}
						nullValueDetails.add(detail);
					}
				}
			}
			//判断数据库中有无该表  有责跳过  无则新建
			dataDisposeDubbo.existTable(httpRequest.getTableName(),jsonStructure(dataModelFieldEntities));


			//保存数据到数据库中
			List<String> listField = stringField(dataModelFieldEntities);
			savaDataToTable(httpRequest.getTableName(),dataArray,listField);


			return;
		} catch (Exception e) {

			XxlJobHelper.log(e);
			XxlJobHelper.handleFail();
			return;
		} finally {
			try {
				if (bufferedReader != null) {
					bufferedReader.close();
				}
				if (connection != null) {
					connection.disconnect();
				}
			} catch (Exception e2) {
				XxlJobHelper.log(e2);
			}
		}
	}

	/**
	 * @author Lion
	 * @description 封装json结构
	 * @param dataModelFieldEntities
	 * @return java.util.List<com.alibaba.fastjson.JSONObject>
	 */
	public List<JSONObject> jsonStructure(List<DataModelFieldEntity> dataModelFieldEntities){

		List<JSONObject> jsonObjects = new ArrayList<>();
		for (int i=0;i< dataModelFieldEntities.size();i++){
			JSONObject jsonObject = new JSONObject();
			jsonObject.put("fieldName",dataModelFieldEntities.get(i).getFieldName());
			jsonObject.put("fieldType",dataModelFieldEntities.get(i).getFieldType());
			jsonObject.put("fieldComments",dataModelFieldEntities.get(i).getFieldCnName());
			jsonObjects.add(jsonObject);
		}
		return jsonObjects;
	}

	/**
	 * @author Lion
	 * @description 获取指定字段
	 * @param  dataModelFieldEntities
	 * @return java.util.List<java.lang.String>
	 */
	public List<String> stringField(List<DataModelFieldEntity> dataModelFieldEntities){

		List<String> jsonObjects = dataModelFieldEntities.stream().map(DataModelFieldEntity::getFieldName).collect(Collectors.toList());;
		return jsonObjects;
	}


	/**
	 * @author Lion
	 * @description 获取指定结构下的数据
	 * @param responseMsg
	 * @param struct
	 * @return com.alibaba.fastjson.JSONArray
	 */
	public JSONArray queryJson(JSONObject responseMsg,String struct){

		JSONArray jsonArray = null;
		JSONObject jsonObject = responseMsg;
		if (struct.contains(",")){
			String[] array = struct.split(",");
			for (int i = 0; i <array.length; i++) {

				if (i == array.length -1){
					jsonArray = jsonObject.getJSONArray(array[i]);
				}else {
					jsonObject = jsonObject.getJSONObject(array[i]);
				}
			}
		}
		return jsonArray;
	}

	/**
	 * @author Lion
	 * @description 单条保存数据到数据库中
	 * @param tableName
	 * @param jsonArray
	 * @param stringList
	 * @return void
	 */
	public void savaDataToTable(String tableName,JSONArray jsonArray,List<String> stringList){

		String sql = "";
		Iterator<Object> iterator = jsonArray.iterator();
		while (iterator.hasNext()){
			JSONObject jsonObject = (JSONObject) iterator.next();
			sql = dataDisposeDubbo.generateInsertStatement(tableName, jsonObject, stringList);
			dataDisposeDubbo.insertDataToTable(sql);
		}

		//执行SQL，保存数据 不创建连接

	}


}

